package cn.pomelo.signal.service.impl;

import cn.pomelo.signal.common.CSVFileUtil;
import cn.pomelo.signal.common.Constants;
import cn.pomelo.signal.common.Util;
import cn.pomelo.signal.config.SignalConfig;
import cn.pomelo.signal.mapper.TradeMapper;
import cn.pomelo.signal.model.Trade;
import cn.pomelo.signal.model.TradeVo;
import cn.pomelo.signal.service.TradeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service("tradeService")
public class TradeServiceImpl implements TradeService {

    @Resource
    private TradeMapper tradeMapper;

    @Resource
    private RedisTemplate<String, List<Trade>> redisTemplate;

    /**
     * 批量插入
     *
     * @param list 要插入的数据列表
     */
    @Override
    @Async("asyncServiceExecutor")
    public void executeAsync(List<Trade> list, CountDownLatch countDownLatch) {
        try {
            System.out.println(Thread.currentThread().getName());
            tradeMapper.batchInsert(list);
        } finally {
            // 很关键, 无论上面程序是否异常必须执行countDown,否则await无法释放
            countDownLatch.countDown();
        }
    }

    /**
     * 插入数据方法一：创建很多个对象给数据库插入
     */

    @Override
    public void getAllData() throws IOException {
        List<File> files = Util.readTxt();
        long start = System.currentTimeMillis();
        List<Trade> allList = new ArrayList<>();
        System.out.println(start + "：开始");
        Long size = 0L;
        for (File file : files) {
            List<Trade> list = new ArrayList<>();
            try (InputStream in = new FileInputStream(file);
                 BufferedReader reader = new BufferedReader(new InputStreamReader(in, SignalConfig.getCharset()))) {
                String s = reader.readLine();
                String[] info = s.split(" ");
                s = reader.readLine();
                // 解析数据
                while (!SignalConfig.getEnd().equals((s = reader.readLine()))) {
                    String[] strings = s.split("\t");
                    Trade trade = new Trade();
                    trade.setCode(info[0]);
                    trade.setName(info[1]);
                    trade.setDate(new Date(strings[0]));
                    trade.setOpen(Double.valueOf(strings[1]));
                    trade.setHighest(Double.valueOf(strings[2]));
                    trade.setLowest(Double.valueOf(strings[3]));
                    trade.setClose(Double.valueOf(strings[4]));
                    trade.setTraded(Long.valueOf(strings[5]));
                    trade.setTransaction(Double.valueOf(strings[6]));
                    list.add(trade);
                    size++;
                }
                allList.addAll(list);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(end + "：读取结束");
        System.out.println("用时：" + (end - start));
        System.out.println(size);
        System.out.println(allList.size());
        // 插入数据库
        List<List<Trade>> lists = Util.splitList(allList, (long) allList.size(), 800000);
        CountDownLatch count = new CountDownLatch(lists.size());
        for (List<Trade> l : lists) {
            executeAsync(l, count);
        }
        try {
            // 线程阻塞，保证所有线程都执行完成，才会走下面的代码，取最后的执行结果
            count.await();
        } catch (InterruptedException e) {
            System.out.println("阻塞异常：" + e.getMessage());
        }
        long wEnd = System.currentTimeMillis();
        System.out.println(end + "：写入数据库结束");
        System.out.println("用时：" + (wEnd - end));
    }

    @Override
    public void readData() throws IOException {
        CSVFileUtil.readToCSV();
        tradeMapper.loadCSV(SignalConfig.getTarget());
    }

    @Override
    public int insert(Trade record) {
        return tradeMapper.insert(record);
    }

    @Override
    public int insertSelective(Trade record) {
        return tradeMapper.insertSelective(record);
    }


    /**
     * 通过日期以及股票代码查询
     *
     * @param d1   开始日期
     * @param d2   结束日期
     * @param code 股票代码
     * @return 符合的所有股票信息
     */
    @Override
    public List<Trade> findAllByDateAndCode(String d1, String d2, String code) {
        return tradeMapper.selectAllByDateAndCode(d1, d2, code);
    }

    /**
     * 查询所有股票信息
     *
     * @return 所有股票信息
     */
    @Override
    public List<Trade> findAllTrades() {
        Logger logger = LoggerFactory.getLogger(getClass());
        logger.info("查询所有股票信息");
        List<Trade> trades = redisTemplate.opsForValue().get(Constants.REDIS_PREFIX);
        if (trades == null) {
            trades = tradeMapper.selectAllTrades();
            redisTemplate.opsForValue().set(Constants.REDIS_PREFIX, trades, 10, TimeUnit.MINUTES);
        }
        logger.info("数量" + trades.size());
        return trades;
    }

    /**
     * 获取连续三天上涨的股票信息
     *
     * @param d 指定某天
     * @return 连续三天上涨的所有股票
     */
    @Override
    public List<TradeVo> getRiseTrade(String d) throws ParseException {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        Date date = d == null ? tradeMapper.maxDate() : format.parse(d);
        Calendar instance = Calendar.getInstance();
        instance.setTime(date);
        instance.add(Calendar.DAY_OF_MONTH, -1);
        // 遇到周六、周日再往前一天
        while (instance.get(Calendar.DAY_OF_WEEK) == Calendar.SUNDAY || instance.get(Calendar.DAY_OF_WEEK) == Calendar.SATURDAY) {
            instance.add(Calendar.DAY_OF_MONTH, -1);
        }
        Date yes = instance.getTime();
        instance.add(Calendar.DAY_OF_MONTH, -1);
        while (instance.get(Calendar.DAY_OF_WEEK) == Calendar.SUNDAY || instance.get(Calendar.DAY_OF_WEEK) == Calendar.SATURDAY) {
            instance.add(Calendar.DAY_OF_MONTH, -1);
        }
        Date twoDaysAgo = instance.getTime();
        return tradeMapper.getRiseTrade(format.format(date), format.format(yes), format.format(twoDaysAgo));
    }

    /**
     * 批量插入
     *
     * @param trades 要插入的集合
     * @return 影响行数
     */
    @Override
    public int batchInsert(List<Trade> trades) {
        return tradeMapper.batchInsert(trades);
    }

}


